package com.tjbklx1.redis.chap06;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.Tuple;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.tjbklx1.redis.JedisUtil;

public class App66_MultiRecipientMessaging {

	public static void main(String[] args) {
		new App66_MultiRecipientMessaging().run();
	}

	private void run() {
		Jedis conn = JedisUtil.getJedisWithDB();
		testMultiRecipientMessaging(conn);
	}

	public void testMultiRecipientMessaging(Jedis conn) {
		System.out.println("\n----- testMultiRecipientMessaging -----");
		conn.del("ids:chat:", "msgs:1", "ids:1", "seen:joe", "seen:jeff", "seen:jenny");

		System.out.println("Let's create a new chat session with some recipients...");
		Set<String> recipients = new HashSet<String>();
		recipients.add("jeff");
		recipients.add("jenny");
		String chatId = createChat(conn, "joe", recipients, "message 1");
		System.out.println("Now let's send a few messages...");
		for (int i = 2; i < 5; i++) {
			sendMessage(conn, chatId, "joe", "message " + i);
		}
		System.out.println();

		System.out.println("And let's get the messages that are waiting for jeff and jenny...");
		List<ChatMessages> r1 = fetchPendingMessages(conn, "jeff");
		List<ChatMessages> r2 = fetchPendingMessages(conn, "jenny");
		System.out.println("They are the same? " + r1.equals(r2));
		assert r1.equals(r2);
		System.out.println("Those messages are:");
		for (ChatMessages chat : r1) {
			System.out.println("  chatId: " + chat.chatId);
			System.out.println("    messages:");
			for (Map<String, Object> message : chat.messages) {
				System.out.println("      " + message);
			}
		}

		conn.del("ids:chat:", "msgs:1", "ids:1", "seen:joe", "seen:jeff", "seen:jenny");
	}

	public static String createChat(Jedis conn, String sender, Set<String> recipients, String message) {
		String chatId = String.valueOf(conn.incr("ids:chat:"));
		return createChat(conn, sender, recipients, message, chatId);
	}

	public static String createChat(Jedis conn, String sender, Set<String> recipients, String message, String chatId) {
		recipients.add(sender);

		Transaction trans = conn.multi();
		for (String recipient : recipients) {
			trans.zadd("chat:" + chatId, 0, recipient);
			trans.zadd("seen:" + recipient, 0, chatId);
		}
		trans.exec();

		return sendMessage(conn, chatId, sender, message);
	}

	/**
	 * 发送消息
	 * @param conn
	 * @param chatId
	 * @param sender
	 * @param message
	 * @return
	 */
	public static String sendMessage(Jedis conn, String chatId, String sender, String message) {
		String identifier = App63_DistributedLock.acquireLock(conn, "chat:" + chatId); //锁
		if (identifier == null) {
			throw new RuntimeException("Couldn't get the lock");
		}
		try {
			long messageId = conn.incr("ids:" + chatId);
			HashMap<String, Object> values = new HashMap<String, Object>();
			values.put("id", messageId);
			values.put("ts", System.currentTimeMillis());
			values.put("sender", sender);
			values.put("message", message);
			String packed = new Gson().toJson(values);
			conn.zadd("msgs:" + chatId, messageId, packed); // 将message 放到zset中
		} finally {
			App63_DistributedLock.releaseLock(conn, "chat:" + chatId, identifier); // 释放锁
		}
		return chatId;
	}

	/**
	 * 取出等待接收的消息
	 * @param conn
	 * @param recipient
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public static List<ChatMessages> fetchPendingMessages(Jedis conn, String recipient) {
		Set<Tuple> seenSet = conn.zrangeWithScores("seen:" + recipient, 0, -1);
		List<Tuple> seenList = new ArrayList<Tuple>(seenSet);

		Transaction trans = conn.multi();
		for (Tuple tuple : seenList) {
			String chatId = tuple.getElement();
			int seenId = (int) tuple.getScore();
			trans.zrangeByScore("msgs:" + chatId, String.valueOf(seenId + 1), "inf");
		}
		List<Object> results = trans.exec();

		Gson gson = new Gson();
		Iterator<Tuple> seenIterator = seenList.iterator();
		Iterator<Object> resultsIterator = results.iterator();

		List<ChatMessages> chatMessages = new ArrayList<ChatMessages>();
		List<Object[]> seenUpdates = new ArrayList<Object[]>();
		List<Object[]> msgRemoves = new ArrayList<Object[]>();
		while (seenIterator.hasNext()) {
			Tuple seen = seenIterator.next();
			Set<String> messageStrings = (Set<String>) resultsIterator.next();
			if (messageStrings.size() == 0) {
				continue;
			}

			int seenId = 0;
			String chatId = seen.getElement();
			List<Map<String, Object>> messages = new ArrayList<Map<String, Object>>();
			for (String messageJson : messageStrings) {
				Map<String, Object> message = (Map<String, Object>) gson.fromJson(messageJson,
						new TypeToken<Map<String, Object>>() {
						}.getType());
				int messageId = ((Double) message.get("id")).intValue();
				if (messageId > seenId) {
					seenId = messageId;
				}
				message.put("id", messageId);
				messages.add(message);
			}

			conn.zadd("chat:" + chatId, seenId, recipient);
			seenUpdates.add(new Object[] { "seen:" + recipient, seenId, chatId });

			Set<Tuple> minIdSet = conn.zrangeWithScores("chat:" + chatId, 0, 0);
			if (minIdSet.size() > 0) {
				msgRemoves.add(new Object[] { "msgs:" + chatId, minIdSet.iterator().next().getScore() });
			}
			chatMessages.add(new ChatMessages(chatId, messages));
		}

		trans = conn.multi();
		for (Object[] seenUpdate : seenUpdates) {
			trans.zadd((String) seenUpdate[0], (Integer) seenUpdate[1], (String) seenUpdate[2]);
		}
		for (Object[] msgRemove : msgRemoves) {
			trans.zremrangeByScore((String) msgRemove[0], 0, ((Double) msgRemove[1]).intValue());
		}
		trans.exec();

		return chatMessages;
	}


}
